package com.dsk.connector.sink.mysql.consumer;


import cn.hutool.json.JSONUtil;
import com.dsk.connector.commons.conf.TableMapping;
import com.dsk.connector.commons.trans.FieldSchema;
import com.dsk.connector.commons.trans.RecordDdl;
import com.dsk.connector.commons.trans.RecordDml;
import com.dsk.connector.commons.trans.TransMsg;
import com.dsk.connector.commons.utils.DDLUtils;
import com.dsk.connector.sink.common.consumer.SinkConsumer;
import com.dsk.connector.sink.common.utils.SinkUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.DependsOn;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.StatementCallback;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

@Component
@DependsOn("tableMappings")
public class MysqlConsumer implements SinkConsumer {
    Logger logger= LoggerFactory.getLogger(MysqlConsumer.class);
    @Autowired
    private NamedParameterJdbcTemplate namedParameterJdbcTemplate;
    @Autowired
    private ApplicationContext context;
    @Resource(name="tableMappings")
    private Map<String, TableMapping> tableMappings;

    private Map<String,String> databaseMapping=null;

    /**
     * 收集源库和目标库的映射关系
     * @return
     */
    public Map<String,String> getMapingDatabases(){
        //
        if(this.databaseMapping!=null){
            return this.databaseMapping;
        }
        //初始化数据
        Map<String,String> mapingDatabases=new HashMap<>();
        for(TableMapping tableMapping:tableMappings.values()){
            if(tableMapping.getSourceTable().contains(".")){
                String sourceDb=tableMapping.getSourceTable().split("\\.")[0];
                String targetDb=tableMapping.getTargetTable().split("\\.")[0];
                mapingDatabases.put(sourceDb,targetDb);
            }else{
                mapingDatabases.put(tableMapping.getSourceTable(),tableMapping.getTargetTable());
            }
        }
        this.databaseMapping=mapingDatabases;
        return mapingDatabases;
    }

    /**
     * 替换ddl中的源表名为目标表名
     * @param tableMapping
     * @param ddl
     * @return
     */
    public String replaceDDlWithTarget(TableMapping tableMapping,String ddl){
        ddl=ddl.toLowerCase();
        Function<String,String> replaceWithDb=(ori)->{
            Map<String,String> dbMappings=this.getMapingDatabases();
            boolean isReplaced=false;
            for(String sourceDb:dbMappings.keySet()){
                if(ori.contains(" "+sourceDb.toLowerCase()+".")){
                    ori=ori.replace(" "+sourceDb.toLowerCase()+"."," "+dbMappings.get(sourceDb)+".");
                }
                if(ori.contains(" `"+sourceDb.toLowerCase()+"`.")){
                    ori=ori.replace(" `"+sourceDb.toLowerCase()+"`."," `"+dbMappings.get(sourceDb)+"`.");
                }
            }
            return ori;
        };
        if(tableMapping.getSourceTable().contains(".")){
            String sourceTableName=tableMapping.getSourceTable().split("\\.")[1].toLowerCase();
            String sourceDb=tableMapping.getSourceTable().split("\\.")[0].toLowerCase();
            String targetDb=tableMapping.getTargetTable().split("\\.")[0];
            String targetTableName=tableMapping.getTargetTable().split("\\.")[1];

            if(ddl.contains(tableMapping.getSourceTable())){
                ddl=ddl.replace(tableMapping.getSourceTable(),tableMapping.getTargetTable());
                //在rename语句中，sourceTable只包括目标表名和库名，也需要把源表的库名替换成目标表的库名
                ddl=replaceWithDb.apply(ddl);
                return ddl;
            }else if(ddl.contains(String.format("`%s`.`%s`",sourceDb,sourceTableName))){
                ddl=ddl.replace(String.format("`%s`.`%s`",sourceDb,sourceTableName),String.format("`%s`.`%s`",targetDb,targetTableName));
                ddl=replaceWithDb.apply(ddl);
                return ddl;
            }else{
                if(ddl.toLowerCase().contains(" "+sourceTableName.toLowerCase())){
                    ddl=ddl.replaceAll("\\s((?i)"+sourceTableName+")(?![a-zA-Z0-9_])",tableMapping.getTargetTable());
                    ddl=replaceWithDb.apply(ddl);
                    return ddl;
                }else{
                    ddl=replaceWithDb.apply(ddl);
                    return "use "+targetDb+";\n"+ddl;
                }
            }
        }else{
            Map<String,String> dbMappings=this.getMapingDatabases();
            String newDdl=replaceWithDb.apply(ddl);
            if(newDdl.equals(ddl)){
                return "use "+dbMappings.get(tableMapping.getSourceTable())+";\n"+ddl;
            }else{
                return newDdl;
            }
        }
    }

    @Override
    public void consume(TransMsg transMsg) throws Exception {
        switch (transMsg.getOperation()){
            case "insert":
                this.handleInsert(transMsg);
                break;
            case "update":
                this.handleUpdate(transMsg);
                break;
            case "delete":
                this.handleDelete(transMsg);
                break;
            case "ddl":
                this.handleDDl(transMsg);
            default:
                break;

        }
    }

    /**
     * 处理删除处理
     * @param transMsg
     */
    private void handleDelete(TransMsg transMsg) {
        TableMapping tableMapping=this.getTableMappings().get(transMsg.getTableName());
        // 如果同步禁止了 insert ，则忽略
        if(!tableMapping.getDMLs().contains("delete")){
            return ;
        }
        StringBuilder whereCond=new StringBuilder();
        whereCond.append("1=1");
        RecordDml recordDml=transMsg.getDatas().get(0);
        for(FieldSchema schema:recordDml.getKeySchemas()){
            whereCond.append(" and ").append(schema.getField()).append("=:"+schema.getField());
        }
        String condStr=whereCond.toString();

        Map<String,List<Map<String,Object>>> targetTableDatas=this.splitTableData(tableMapping,transMsg.getDatas(),false,null);
        targetTableDatas.forEach((targetTable,datas)->{
            //删除数据
            String deleteSql=String.format("delete from  %s  where %s", targetTable, condStr);
            this.namedParameterJdbcTemplate.batchUpdate(deleteSql,datas.toArray(new HashMap[0]));
        });



    }

    private void handleUpdate(TransMsg transMsg) {
        TableMapping tableMapping=this.getTableMappings().get(transMsg.getTableName());
        // 如果同步禁止了 insert ，则忽略
        if(!tableMapping.getDMLs().contains("update")){
            return ;
        }
        StringBuilder whereCond=new StringBuilder();
        whereCond.append("1=1");
        for(FieldSchema schema:transMsg.getDatas().get(0).getKeySchemas()){
            whereCond.append(" and ").append(schema.getField()).append("=:key"+schema.getField()+" ");
        }
        String condStr=whereCond.toString();

        StringBuilder setter = new StringBuilder();
        for(String f:transMsg.getModifyFields().split(",")){
            //由于无主键的表，需要全字段做为查询条件，所以不能排除任何字段
//            if(condStr.contains("=:key"+f+" ")){
//                continue;
//            }
            //
            setter.append(",").append(f+"=:"+f);
        }
        String setterStr=setter.toString().substring(1);

        Map<String,List<Map<String,Object>>> targetTableDatas=this.splitTableData(tableMapping,transMsg.getDatas(),true,"key");
        targetTableDatas.forEach((targetTable,datas)->{
            //创建表
            if(tableMapping.isSplitTable()){
                this.createTargetTable(targetTable,tableMapping.getTargetTable());
            }
            //更新数据
            String updateSql=String.format("update %s set %s where %s", targetTable, setterStr, condStr);

            this.namedParameterJdbcTemplate.batchUpdate(updateSql,datas.toArray(new HashMap[0]));
        });
    }

    private void handleDDl(TransMsg transMsg) {
        if(transMsg.getDatas()==null || transMsg.getDatas().size()==0){
            String objStr=JSONUtil.toJsonStr(transMsg);
            System.out.println(objStr);
            logger.error("消费错误，消息对象没有datas属性："+objStr);
            return ;
        }
        String ddlSql=transMsg.getFirstDDL();
        for(RecordDml recordDml:transMsg.getDatas()){

            //替换表名
            TableMapping tableMapping=this.getTableMappings().get(recordDml.getTableName().toLowerCase());
            if(tableMapping==null && recordDml.getTableName().contains(".")){
                tableMapping=this.tableMappings.get(recordDml.getTableName().split("\\.")[0]);
            }
            if(tableMapping==null){
                //记录ddl语句
                this.insertDdlRecord(ddlSql,recordDml.getTableName(),recordDml.getOperation(),false,transMsg.getSourcePoint());
                //如果影响更新的里面，不包括配置表，则忽略不执行
                return;
            }
            if(!tableMapping.getDDLs().contains(recordDml.getOperation())){
                //记录ddl语句
                this.insertDdlRecord(ddlSql,recordDml.getTableName(),recordDml.getOperation(),false,transMsg.getSourcePoint());
                //如果未配置表的alter规则，则忽略执行
                return;
            }
            //记录ddl语句
            this.insertDdlRecord(ddlSql,recordDml.getTableName(),recordDml.getOperation(),true,transMsg.getSourcePoint());
            //替换语句的表为目标表
            ddlSql=this.replaceDDlWithTarget(tableMapping,ddlSql);
        }
        //去除语句中的注释行，防止执行出错
        final String ddlSqlFinal= DDLUtils.removeComment(ddlSql);
        //执行alter语句，担心语句中有use 语句切换数据库，因此需要使用StatementCallback处理
        namedParameterJdbcTemplate.getJdbcTemplate().execute(new StatementCallback<Object>() {
            @Override
            public Object doInStatement(Statement stmt) throws SQLException, DataAccessException {
                //查询正在使用的数据库
                ResultSet rs=stmt.executeQuery(" select database()");
                String database=null;
                if(rs.next()){
                    database=rs.getString(1);
                }
                rs.close();
                try{
                    //执行sql语句
                    if(ddlSqlFinal.matches(".+;\\s*\r?\n.*")){
                        String[] sqls=ddlSqlFinal.split(";\\s*\r?\n");
                        for(String sql:sqls){
                            if(!StringUtils.isBlank(sql)){
                                stmt.execute(sql);
                            }
                        }
                    }else{
                        stmt.execute(ddlSqlFinal);
                    }
                }finally {
                    //切换到原来的数据库
                    if(StringUtils.isNotBlank(database)){
                        stmt.execute("use "+database);
                    }
                }
                return 1;
            }
        }) ;
    }

    /**
     * 保存ddl的操作历史记录
     * @param ddlSql
     */
    private void insertDdlRecord(String ddlSql,String tableName,String operation,boolean isExcuted,Map<String,Object> sourcePoint) {
        //获取系统开关
        String recordDDL=this.context.getEnvironment().getProperty("config.sink.recordDDL");
        String ddlTableName=this.context.getEnvironment().getProperty("config.sink.ddl_tablename");
        if(!"true".equalsIgnoreCase(recordDDL)){
            return ;
        }
        if(StringUtils.isBlank(ddlTableName)){
            ddlTableName="datatrans_ddl_record";
        }

        //创建表
        String createTableSql="CREATE TABLE if not exists datatrans_ddl_record (\n" +
                "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                "  `table_name` varchar(255) DEFAULT NULL,\n" +
                "  `vc_operation` varchar(100) DEFAULT NULL,"+
                "  `operate_time` datetime default null,"+
                "  `is_executed` char(1) DEFAULT NULL,\n"+
                "  `ddl_sql` text DEFAULT NULL,\n" +
                "  `create_time` timestamp NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),\n" +
                "  PRIMARY KEY (`id`)\n" +
                ") ENGINE=InnoDB DEFAULT CHARSET=utf8";
        //替换为配置的表名
        createTableSql=createTableSql.replace("datatrans_ddl_record",ddlTableName);

        this.namedParameterJdbcTemplate.getJdbcTemplate().execute(createTableSql);
        //插入记录
        String insertSql="insert into datatrans_ddl_record(table_name,vc_operation,operate_time,is_executed,ddl_sql) values(:tableName,:vc_operation,:operate_time,:isExecuted,:ddlSql)";
        //替换为配置的表名
        insertSql=insertSql.replace("datatrans_ddl_record",ddlTableName);

        Map<String,Object> param=new HashMap<>();
        param.put("tableName",tableName);
        param.put("ddlSql",ddlSql);
        param.put("isExecuted",isExcuted?1:0);
        param.put("vc_operation",operation);
        param.put("operate_time",sourcePoint.get("ts_ms"));
        this.namedParameterJdbcTemplate.update(insertSql,param);
    }

    public Map<String, TableMapping> getTableMappings() {
        if(tableMappings==null){
            tableMappings=(Map<String, TableMapping>)context.getBean("tableMappings");
        }
        return tableMappings;
    }

    protected void handleInsert(TransMsg transMsg) {
        TableMapping tableMapping=this.getTableMappings().get(transMsg.getTableName());
        // 如果同步禁止了 insert ，则忽略
        if(!tableMapping.getDMLs().contains("insert")){
            return ;
        }
        StringBuilder holder = new StringBuilder();
        for(String f:transMsg.getModifyFields().split(",")){
            holder.append(",").append(":").append(f);
        }
        String replaceHolder=holder.toString().substring(1);
        //如果是不拆分表，按原名表返回
        Map<String,List<Map<String,Object>>> targetTableDatas=this.splitTableData(tableMapping,transMsg.getDatas(),false,null);
        targetTableDatas.forEach((targetTable,datas)->{
            //创建表
            if(tableMapping.isSplitTable()){
                this.createTargetTable(targetTable,tableMapping.getTargetTable());
            }
            //插入数据
            String insertSql=String.format("replace into %s (%s) values (%s)", targetTable, transMsg.getModifyFields(), replaceHolder);
            this.namedParameterJdbcTemplate.batchUpdate(insertSql,datas.toArray(new HashMap[0]));
        });

    }

    /**
     * 创建目标表
     * @param targetNewTable 需要创建的目标表
     * @param targetOrigTable 原表
     */
    private void createTargetTable(String targetNewTable, String targetOrigTable) {
        String createTableSql=String.format("create table if not exists %s like %s",targetNewTable,targetOrigTable);
        this.namedParameterJdbcTemplate.getJdbcTemplate().execute(createTableSql);
    }

    /**
     * 根据分隔规则分隔数据到不同的表中
     * @param tableMapping
     * @param datas
     * @return
     */
    private Map<String, List<Map<String, Object>>> splitTableData(TableMapping tableMapping, List<RecordDml> datas,boolean containKeyValue,String keyPrefix) {
        Map<String, List<Map<String, Object>>> targetTableDatas=new HashMap<>();
        datas.forEach(record->{
            String targetTable= SinkUtils.getTargetTable(tableMapping,record);
            Map<String,Object> values=new HashMap<>();
            values.putAll(record.getPayloadValues());
            if(containKeyValue){
                record.getKeyValue().forEach((k,v)->{
                    values.put(keyPrefix+k,v);
                });
            }
            if(targetTableDatas.containsKey(targetTable)){
                targetTableDatas.get(targetTable).add(values);
            }else{
                List<Map<String,Object>> list=new LinkedList<>();
                list.add(values);
                targetTableDatas.put(targetTable,list);
            }
        });

        return targetTableDatas;
    }
}
